package com.tang.aggregate;

import com.tang.bean.WaterSensor;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * 简单聚合
 *
 * @author tang
 * @since 2023/6/7 12:27
 */
public class SimpleAggregateDemo {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        DataStreamSource<WaterSensor> sensorDataStreamSource = env.fromElements(
                new WaterSensor("s1", 1L, 1),
                new WaterSensor("s1", 11L, 11),
                new WaterSensor("s2", 2L, 2),
                new WaterSensor("s3", 3L, 3)
        );

        KeyedStream<WaterSensor, String> sensorKeyedStream = sensorDataStreamSource.keyBy(new KeySelector<WaterSensor, String>() {
            @Override
            public String getKey(WaterSensor value) throws Exception {
                return value.getId();
            }
        });

        /*
         * 1、 keyby之后才能调用
         * 2、分组内的聚合：对同一个key的数据进行聚合
         */
        // Cannot reference field by position on PojoType<com.tang.bean.WaterSensor .... 传位置索引的，适用于 Tuple类型，POJO不行
        //SingleOutputStreamOperator<WaterSensor> result = sensorKeyedStream.sum(2);
        //SingleOutputStreamOperator<WaterSensor> result = sensorKeyedStream.sum("vc");
        /*
         * WaterSensor{id='s1', ts=1, vc=1}
         * WaterSensor{id='s1', ts=1, vc=12}
         * WaterSensor{id='s2', ts=2, vc=2}
         * WaterSensor{id='s3', ts=3, vc=3}
         */

        /*
         * max\maxby的区别： 同min
         * - max：只会取比较字段的最大值，非比较字段保留第一次的值
         * - maxby：取比较字段的最大值，同时非比较字段 取 最大值这条数据的值
         */
        SingleOutputStreamOperator<WaterSensor> result = sensorKeyedStream.max("vc");
        /*
         * WaterSensor{id='s1', ts=1, vc=1}
         * WaterSensor{id='s1', ts=1, vc=11}
         * WaterSensor{id='s2', ts=2, vc=2}
         * WaterSensor{id='s3', ts=3, vc=3}
         */
        //SingleOutputStreamOperator<WaterSensor> result = sensorKeyedStream.min("vc");
        /*
         * WaterSensor{id='s1', ts=1, vc=1}
         * WaterSensor{id='s1', ts=1, vc=1}
         * WaterSensor{id='s2', ts=2, vc=2}
         * WaterSensor{id='s3', ts=3, vc=3}
         */
        //SingleOutputStreamOperator<WaterSensor> result = sensorKeyedStream.maxBy("vc");
        /*
         * WaterSensor{id='s1', ts=1, vc=1}
         * WaterSensor{id='s1', ts=11, vc=11}
         * WaterSensor{id='s2', ts=2, vc=2}
         * WaterSensor{id='s3', ts=3, vc=3}
         */
        //SingleOutputStreamOperator<WaterSensor> result = sensorKeyedStream.minBy("vc");
        /*
         * WaterSensor{id='s1', ts=1, vc=1}
         * WaterSensor{id='s1', ts=1, vc=1}
         * WaterSensor{id='s2', ts=2, vc=2}
         * WaterSensor{id='s3', ts=3, vc=3}
         */

        result.print();

        env.execute();
    }

}
